package com.hub.entity;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.HashMap;
import java.util.Random;

public class MySensor  implements SourceFunction<SensorReading>{

    private boolean runningStatus = true;

    private int sensorNum = 1;
    @Override
    public void run(SourceFunction.SourceContext<SensorReading> sourceContext) throws Exception {

        Random random = new Random();
        HashMap<String, Double> sensorTempMap = new HashMap<String, Double>();
        for( int i = 0; i < sensorNum; i++ ){
//            double numTemp = random.nextInt(10);
            double numTemp = 60 + random.nextGaussian() * 20; // 高斯分布
            sensorTempMap.put("sensor_" + (i + 1), numTemp);
        }

        while(runningStatus){
            for( String sensorId: sensorTempMap.keySet() ){
//                Double newTemp = sensorTempMap.get(sensorId) + random.nextGaussian(); sensorTempMap.put(sensorId, newTemp);
                double newTemp = random.nextInt(10);;
                sourceContext.collect( new SensorReading(sensorId, System.currentTimeMillis(),newTemp));
            }
            Thread.sleep(1000L);
        }
    }

    @Override
    public void cancel() {
        runningStatus = false;
    }
}
